import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.text.ParseException;
import java.util.*;
import database.EventSerializer;
import optim.BatchSizeCalculator;
import main.*;
import event.*;
import event.DEFINES;
import streamdef.*;
import streamproc.buffercomponents.Buffer;
import streamproc.buffercomponents.BufferManagerComponent;

public class Main {
	private static boolean usePostgre = false;
	private static String mySQLDBURL = "jdbc:mysql://localhost:3306/CEPL";
	private static String mySQLDBUser = "root";
	private static String mySQLDBPassword = "cepl";

	private static String postgreDBURL = "jdbc:postgresql://localhost:5432/CEPL";
	private static String postgreDBUser = "cepl";
	private static String postgreDBPassword = "cepl";

	private static class PrintObject implements EventCallbackFunctionOwner {
		public PrintObject(String s) {
			stringToPrint = s;
		}

		private String stringToPrint = null;

		@Override
		public void callFunctionForNewComplexEvent(OutgoingComplexEventInfo ocei) {
			System.out.println(stringToPrint + " --> streamID = "
					+ ocei.streamID + ", eventID = " + ocei.eventID
					+ ", timestamp = " + ocei.timestamp);
		}

		@Override
		public void callFunctionForVectorOfNewComplexEvents(
				Vector<OutgoingComplexEventInfo> events) {
			for (int i = 0; i < events.size(); ++i) {
				OutgoingComplexEventInfo ocei = events.get(i);
				System.out.println(stringToPrint + " --> streamID = "
						+ ocei.streamID + ", eventID = " + ocei.eventID
						+ ", timestamp = " + ocei.timestamp);
			}
		}
	}

	public static void main(String args[]) {
		if (usePostgre == false)
			CEPLFunctions.Main.init(usePostgre, !usePostgre, mySQLDBURL,
					mySQLDBUser, mySQLDBPassword);
		else
			CEPLFunctions.Main.init(usePostgre, !usePostgre, postgreDBURL,
					postgreDBUser, postgreDBPassword);

		// testCallbacks();
		// testEventMessageFormats();
		// testEventMessageFormats2();

		// EventSerializer es = EventSerializer.instance();
		// es.dropTable();
		// es.createTable();

		// testEventsSerialization();
		// testEventsDeserialization();

		// testBufferCleanup();

		// testPreFiltering();

		testSequenceDetector();

		// fileGenerator();
	}

	public static void testCallbacks() {
		Calendar calendar = Calendar.getInstance();
		java.util.Date now = calendar.getTime();
		java.sql.Timestamp currentTimestamp = new java.sql.Timestamp(
				now.getTime());

		PrintObject po1 = new PrintObject(new String("print1"));
		PrintObject po2 = new PrintObject(new String("print2"));
		PrintObject po3 = new PrintObject(new String("print3"));
		PrintObject po4 = new PrintObject(new String("print4"));

		CEPLFunctions.EventCallbacks.addEventCallback(1, po1);
		CEPLFunctions.EventCallbacks.addEventCallback(1, po2);
		CEPLFunctions.EventCallbacks.addEventCallback(2, po2);
		CEPLFunctions.EventCallbacks.addEventCallback(2, po3);
		CEPLFunctions.EventCallbacks.addEventCallback(3, po4);

		byte[] temp = new byte[0];
		CEPLFunctions.EventCallbacks.fireEvent(new Event(1, false, true, true,
				temp, currentTimestamp));
		CEPLFunctions.EventCallbacks.fireEvent(new Event(2, false, true, true,
				temp, currentTimestamp));
		CEPLFunctions.EventCallbacks.fireEvent(new Event(3, false, true, true,
				temp, currentTimestamp));
		System.out.println();
		CEPLFunctions.EventCallbacks.removeEventCallback(1, po2);
		CEPLFunctions.EventCallbacks.fireEvent(new Event(1, false, true, true,
				temp, currentTimestamp));
		CEPLFunctions.EventCallbacks.fireEvent(new Event(2, false, true, true,
				temp, currentTimestamp));
	}

	@SuppressWarnings("deprecation")
	public static void testEventMessageFormats() {
		Vector<DEFINES.CEPLEventDataFieldDefinition> format = new Vector<DEFINES.CEPLEventDataFieldDefinition>();

		// double
		DEFINES.CEPLEventDataFieldDefinition def1 = new DEFINES.CEPLEventDataFieldDefinition();
		def1.fieldName = new String("field1");
		def1.isNumerical = true;
		def1.isFloatingPoint = true;
		def1.isPrecededByLengthInBytes = false;
		def1.lengthInBytes = 8;
		format.add(def1);

		// string with preceding length
		DEFINES.CEPLEventDataFieldDefinition def2 = new DEFINES.CEPLEventDataFieldDefinition();
		def2.fieldName = new String("field2");
		def2.isNumerical = false;
		def2.isPrecededByLengthInBytes = true;
		def2.lengthInBytes = 0;
		format.add(def2);

		// string
		DEFINES.CEPLEventDataFieldDefinition def3 = new DEFINES.CEPLEventDataFieldDefinition();
		def3.fieldName = new String("field3");
		def3.isNumerical = false;
		def3.isPrecededByLengthInBytes = false;
		def3.lengthInBytes = 2;
		format.add(def3);

		// unsigned short
		DEFINES.CEPLEventDataFieldDefinition def4 = new DEFINES.CEPLEventDataFieldDefinition();
		def4.fieldName = new String("field4");
		def4.isNumerical = true;
		def4.isFloatingPoint = false;
		def4.isPrecededByLengthInBytes = false;
		def4.lengthInBytes = 2;
		format.add(def4);

		CEPLFunctions.Events.addEventFormat(1, format);

		double f1 = 5.0;
		short sizef2 = 4;
		String f2 = new String("haha");
		String f3 = new String("bi");
		short f4 = 9;

		ByteBuffer buf = ByteBuffer.allocate(18);
		buf.putDouble(f1);
		buf.putShort(sizef2);
		buf.put(f2.getBytes());
		buf.put(f3.getBytes());
		buf.putShort(f4);

		byte[] data = buf.array();

		HashMap<String, event.DEFINES.CEPLEventDataFieldValue> testMap = main.CEPLFunctions.Events
				.parseEvent(1, data);
		Iterator<Map.Entry<String, event.DEFINES.CEPLEventDataFieldValue>> iterator = testMap
				.entrySet().iterator();
		while (iterator.hasNext()) {
			Map.Entry<String, event.DEFINES.CEPLEventDataFieldValue> me = (Map.Entry<String, event.DEFINES.CEPLEventDataFieldValue>) iterator
					.next();

			System.out.print(me.getKey() + " => ");
			if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.CHAR)
				System.out.println((Character) (me.getValue().value));
			if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.DOUBLE)
				System.out.println((Double) (me.getValue().value));
			if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.FLOAT)
				System.out.println((Float) (me.getValue().value));
			if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.INT)
				System.out.println((Integer) (me.getValue().value));
			if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.SHORT)
				System.out.println((Short) (me.getValue().value));
			if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.UNKNOWN)
				System.out.println("UNKNOWN");
			if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.TEXT)
				System.out.println((String) (me.getValue().value));
		}
	}

	public static void testEventMessageFormats2() {
		try {
			String query = "create pstream(1,field1:double,field2:text:0,field3:text:2, field4:short)";
			StreamDefinitionParser.instance().parse(query);

			String query2 = "create pstream(2,fielda:double,fieldb:text:0,fieldc:text:2, fieldd:short)";
			StreamDefinitionParser.instance().parse(query2);

			String query3 = "create cstream(3,test1:1.field2,test2:1.field1,test3:2.fieldd)\n";
			query3 += "from 1:1[200],2:2[100],3:1[!100]\n";
			query3 += "where 1.field1 == 2\n";
			query3 += "And (2.fielda <= 1.field4 and 1.field2 = 'test')\n";
			query3 += "and 2 before 1";
			StreamDefinitionParser.instance().parse(query3);

			double f1 = 5.0;
			short sizef2 = 4;
			String f2 = new String("haha");
			String f3 = new String("bi");
			short f4 = 9;

			ByteBuffer buf = ByteBuffer.allocate(18);
			buf.putDouble(f1);
			buf.putShort(sizef2);
			buf.put(f2.getBytes());
			buf.put(f3.getBytes());
			buf.putShort(f4);

			byte[] eventData = buf.array();

			HashMap<String, event.DEFINES.CEPLEventDataFieldValue> testMap = main.CEPLFunctions.Events
					.parseEvent(1, eventData);
			Iterator<Map.Entry<String, event.DEFINES.CEPLEventDataFieldValue>> iterator = testMap
					.entrySet().iterator();
			while (iterator.hasNext()) {
				Map.Entry<String, event.DEFINES.CEPLEventDataFieldValue> me = (Map.Entry<String, event.DEFINES.CEPLEventDataFieldValue>) iterator
						.next();

				System.out.print(me.getKey() + " => ");
				if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.CHAR)
					System.out.println((Character) (me.getValue().value));
				if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.DOUBLE)
					System.out.println((Double) (me.getValue().value));
				if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.FLOAT)
					System.out.println((Float) (me.getValue().value));
				if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.INT)
					System.out.println((Integer) (me.getValue().value));
				if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.SHORT)
					System.out.println((Short) (me.getValue().value));
				if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.UNKNOWN)
					System.out.println("UNKNOWN");
				if (me.getValue().type == event.DEFINES.CEPLEventDataFieldType.TEXT)
					System.out.println((String) (me.getValue().value));
			}
		} catch (ParseException e) {
			e.printStackTrace();
		}
	}

	public static void testEventsSerialization() {
		try {
			String query = "create pstream(1,field1:double,field2:text:0,field3:text:2, field4:short)";
			StreamDefinitionParser.instance().parse(query);

			double f1 = 5.0;
			short sizef2 = 4;
			String f2 = new String("haha");
			String f3 = new String("bi");
			short f4 = 9;

			ByteBuffer buf = ByteBuffer.allocate(25);
			for (int i = 0; i < 1; ++i) {
				// sum of lengths = 18 bytes long
				buf.putDouble(f1);
				buf.putShort(sizef2);
				buf.put(f2.getBytes());
				buf.put(f3.getBytes());
				buf.putShort(f4);
			}

			ByteBuffer buf2 = ByteBuffer.allocate(25 - buf.remaining());
			buf2.put(buf.array(), 0, buf2.capacity());
			byte[] data = buf2.array();

			BatchSizeCalculator.instance().doTest(70, 150, 15, 1, data,
					"test.csv");
		} catch (ParseException e) {
			e.printStackTrace();
		}

		// 1000 events to store, autocommit -> 1.3 seconds
		// 1000 events to store, no autocommit -> 1.5 seconds
		// 1000 events to store, commit per 100 -> 0.85 seconds
		// 100000 events, commit per 500 -> 40.979 seconds
		// 100000 events, commit per 5000 -> 41.606 seconds
		// 100000 events, commit per 500 (with prim key) -> 46.374 seconds
		// 100000 events, autocommit -> 82.376 seconds
		// 100000 events, use one _long_ string per 500 -> 19.377s
		// 100000 events, use one _long_ string per 100 -> 11.5s
	}

	public static void testEventsDeserialization() {
		try {
			String query = "create pstream(1,field1:double,field2:text:0,field3:text:2, field4:short)";
			StreamDefinitionParser.instance().parse(query);
			EventSerializer es = EventSerializer.instance();
			Event e = es.retrieveEvent(es.getMaximumSerializedID());

			System.out.println("ID = " + e.getID());
			System.out.println("streamID = " + e.getStreamID());
			System.out.println("timestamp = " + e.getTimestamp());
			System.out.println("isCacheable = " + e.isCacheable());
			System.out.println("isPrimitive = " + e.isPrimitive());
			System.out.println("isForwarded = " + e.isForwarded());
			System.out.println("field1 = " + e.getDouble("field1"));
			System.out.println("field2 = " + e.getString("field2"));
			System.out.println("field3 = " + e.getString("field3"));
			System.out.println("field4 = " + e.getShort("field4"));
		} catch (ParseException e) {
			e.printStackTrace();
		} catch (UnknownDataFieldException e) {
			e.printStackTrace();
		}
	}

	public static void testBufferCleanup() {
		try {
			CEPLFunctions.Main.reset();

			String query = "create pstream(1,field1:double,field2:text:0,field3:text:2, field4:short)";
			StreamDefinitionParser.instance().parse(query);

			String query3 = "create cstream(2,test1:1.field2)\n";
			query3 += "from 1:1[1.0]\n"; // 1 second buffer
			query3 += "where 1.field1 == 2\n";
			StreamDefinitionParser.instance().parse(query3);

			CEPLFunctions.Main.start();

			double f1 = 5.0;
			short sizef2 = 4;
			String f2 = new String("haha");
			String f3 = new String("bi");
			short f4 = 9;

			ByteBuffer buf = ByteBuffer.allocate(18);
			buf.putDouble(f1);
			buf.putShort(sizef2);
			buf.put(f2.getBytes());
			buf.put(f3.getBytes());
			buf.putShort(f4);

			byte[] data = buf.array();

			Calendar calendar = Calendar.getInstance();
			java.util.Date now = calendar.getTime();
			java.sql.Timestamp prevTimestamp = new java.sql.Timestamp(
					now.getTime());

			for (int i = 0; i < 500000; ++i) {
				main.CEPLFunctions.Events.pushEvent(1, data);
			}

			calendar = Calendar.getInstance();
			now = calendar.getTime();
			java.sql.Timestamp currentTimestamp = new java.sql.Timestamp(
					now.getTime());
			double diff = (currentTimestamp.getTime() - prevTimestamp.getTime()) / 1000.0;
			double eventsPerSec = 500000.0 / diff;
			System.out.println(eventsPerSec + " events per second pushed");

			Thread.sleep(1);

			calendar = Calendar.getInstance();
			now = calendar.getTime();
			currentTimestamp = new java.sql.Timestamp(now.getTime());

			Buffer b = BufferManagerComponent.instance().getBuffer(1);

			if (!b.checkValidnessOrder())
				System.out.println("invalid order!");
			b.removeOutdatedEventsBinarySearch(currentTimestamp);

			System.out.println("currentTimestamp = " + currentTimestamp);
			System.out.println("oldestEvent = "
					+ b.getOldestEvent().getTimestamp());
			System.out.println("newestEvent = "
					+ b.getNewestEvent().getTimestamp());
		} catch (ParseException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public static void testPreFiltering() {
		try {
			CEPLFunctions.Main.reset();

			String primQuery1 = "create pstream(1,field1:double,field2:text:0,field3:text:2, field4:short)";
			StreamDefinitionParser.instance().parse(primQuery1);

			String primQuery2 = "create pstream(2,fieldx:short,fieldy:text:0,fieldz:double)";
			StreamDefinitionParser.instance().parse(primQuery2);

			String complexQuery1 = "create cstream(3,test1:1.field2,test2:2.fieldz )\n";
			complexQuery1 += "from 1:1[1.0],2:2[2.0]\n"; // 1 second buffer
			complexQuery1 += "where 1.field1 <= 2\n";
			complexQuery1 += "and 2.fieldx <= 5\n";
			StreamDefinitionParser.instance().parse(complexQuery1);

			String complexQuery2 = "create cstream(4,test55:2.field1)\n";
			complexQuery2 += "from 2:1[1.0]\n"; // 1 second buffer
			complexQuery2 += "where 2.field1 >= 10\n";
			complexQuery2 += "and 2.field2 == 'test'\n";
			StreamDefinitionParser.instance().parse(complexQuery2);

			String complexQuery3 = "create cstream(5,test55:2.field1)\n";
			complexQuery3 += "from 2:1[1.0],3:3[5.0] \n"; // 1 second buffer
			complexQuery3 += "where 2 BEFORE 3\n";
			StreamDefinitionParser.instance().parse(complexQuery3);

			CEPLFunctions.Main.start();

			double f1 = 2.0;
			short sizef2 = 4;
			String f2 = new String("haha");
			String f3 = new String("bi");
			short f4 = 9;

			ByteBuffer buf = ByteBuffer.allocate(18);
			buf.putDouble(f1);
			buf.putShort(sizef2);
			buf.put(f2.getBytes());
			buf.put(f3.getBytes());
			buf.putShort(f4);

			byte[] data = buf.array();

			Calendar calendar = Calendar.getInstance();
			java.util.Date now = calendar.getTime();
			java.sql.Timestamp prevTimestamp = new java.sql.Timestamp(
					now.getTime());

			for (int i = 0; i < 500000; ++i) {
				main.CEPLFunctions.Events.pushEvent(1, data);
			}

			calendar = Calendar.getInstance();
			now = calendar.getTime();
			java.sql.Timestamp currentTimestamp = new java.sql.Timestamp(
					now.getTime());
			double diff = (currentTimestamp.getTime() - prevTimestamp.getTime()) / 1000.0;
			double eventsPerSec = 500000.0 / diff;
			System.out.println(eventsPerSec + " events per second pushed");

			Thread.sleep(1);

			calendar = Calendar.getInstance();
			now = calendar.getTime();
			currentTimestamp = new java.sql.Timestamp(now.getTime());

			Buffer b = BufferManagerComponent.instance().getBuffer(1);

			if (!b.checkValidnessOrder())
				System.out.println("invalid order!");
			b.removeOutdatedEventsBinarySearch(currentTimestamp);

			System.out.println("currentTimestamp = " + currentTimestamp);
			System.out.println("oldestEvent = "
					+ b.getOldestEvent().getTimestamp());
			System.out.println("newestEvent = "
					+ b.getNewestEvent().getTimestamp());
		} catch (ParseException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public static void testSequenceDetector() {
		try {
			CEPLFunctions.Main.reset();

			String primQuery1 = "create pstream(1,field1:double,field2:text:0,field3:text:2, field4:short)";
			StreamDefinitionParser.instance().parse(primQuery1);

			String primQuery2 = "create pstream(2,fieldx:short)";
			StreamDefinitionParser.instance().parse(primQuery2);

			// String complexQuery1 =
			// "create cstream(3,test1:1.field2,test2:2.fieldx )\n";
			// complexQuery1 += "from 1:1[1.0],2:2[2.0],3:2[!0.1]\n";
			// complexQuery1 += "where 1 BEFORE 2\n";
			// complexQuery1 += "and 2.fieldx <= 10\n";
			// complexQuery1 += "and 3.fieldx = 5\n";
			// QueryParser.instance().parse(complexQuery1);

			String complexQuery1 = "create cstream(3,test1:1.field2 )\n";
			complexQuery1 += "from 1:1[2.0],2:2[!1.0]\n";
			complexQuery1 += "WHere 1.field1 == 2.0\n";
			complexQuery1 += "and 2.fieldx == 8\n";
			StreamDefinitionParser.instance().parse(complexQuery1);

			PrintObject po3 = new PrintObject(new String("print3"));
			CEPLFunctions.EventCallbacks.addEventCallback(3, po3);

			CEPLFunctions.Main.start();

			//
			// event 1
			//
			double f1 = 2.0;
			short sizef2 = 4;
			String f2 = new String("haha");
			String f3 = new String("bi");
			short f4 = 9;

			ByteBuffer buf = ByteBuffer.allocate(18);
			buf.putDouble(f1);
			buf.putShort(sizef2);
			buf.put(f2.getBytes());
			buf.put(f3.getBytes());
			buf.putShort(f4);

			byte[] data = buf.array();

			//
			// event 2
			//
			short _f1 = 5;

			ByteBuffer _buf = ByteBuffer.allocate(2);
			_buf.putShort(_f1);

			byte[] _data = _buf.array();

			//
			// event 3
			//
			short __f1 = 15;

			ByteBuffer __buf = ByteBuffer.allocate(2);
			__buf.putShort(__f1);

			byte[] __data = __buf.array();

			//
			// event 4
			//
			short ___f1 = 8;

			ByteBuffer ___buf = ByteBuffer.allocate(2);
			___buf.putShort(___f1);

			byte[] ___data = ___buf.array();

			Calendar calendar = Calendar.getInstance();
			java.util.Date now = calendar.getTime();
			java.sql.Timestamp prevTimestamp = new java.sql.Timestamp(
					now.getTime());

			long teller = 0;
			main.CEPLFunctions.Events.pushEvent(2, ___data);
			for (int i = 0; i < 150000; ++i) {
				if (i % 50 == 0) {
					teller++;
					main.CEPLFunctions.Events.pushEvent(1, data);
				}
				main.CEPLFunctions.Events.pushEvent(2, __data);
				main.CEPLFunctions.Events.pushEvent(2, __data);
				main.CEPLFunctions.Events.pushEvent(2, __data);
				if (i % 13 == 0) {
					teller++;
					main.CEPLFunctions.Events.pushEvent(2, _data);
				}
				main.CEPLFunctions.Events.pushEvent(2, __data);
				teller += 4;
			}

			calendar = Calendar.getInstance();
			now = calendar.getTime();
			java.sql.Timestamp currentTimestamp = new java.sql.Timestamp(
					now.getTime());
			double diff = (currentTimestamp.getTime() - prevTimestamp.getTime()) / 1000.0;
			double eventsPerSec = teller / diff;
			System.out.println(eventsPerSec + " events per second pushed");

			Thread.sleep(1);

			calendar = Calendar.getInstance();
			now = calendar.getTime();
			currentTimestamp = new java.sql.Timestamp(now.getTime());

			Buffer b = BufferManagerComponent.instance().getBuffer(2);

			if (!b.checkValidnessOrder())
				System.out.println("invalid order!");
			b.removeOutdatedEventsBinarySearch(currentTimestamp);

			System.out.println("currentTimestamp = " + currentTimestamp);
			System.out.println("oldestEvent = "
					+ b.getOldestEvent().getTimestamp());
			System.out.println("newestEvent = "
					+ b.getNewestEvent().getTimestamp());
		} catch (ParseException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	@SuppressWarnings("rawtypes")
	public static String dumpInfo() {
		StringBuffer sb = new StringBuffer();
		Runtime rt = Runtime.getRuntime();

		long freeMemory = rt.freeMemory();
		long totalMemory = rt.totalMemory();

		sb.append("free memory=" + freeMemory);
		sb.append("\n");
		sb.append("total memory=" + totalMemory);
		sb.append("\n");

		java.util.Properties p = null;

		try {
			p = System.getProperties();
		} catch (Exception e) {
			e.printStackTrace();
			return "";
		}

		java.util.Enumeration en = p.propertyNames();

		while (en.hasMoreElements()) {
			String s = (String) en.nextElement();
			String strValue = p.getProperty(s);
			sb.append(s + "=<" + strValue + ">");
			sb.append("\n");
		}

		return sb.toString();
	}

	public static void fileGenerator() {
		String folder = "V:\\Dropbox\\Business Plan\\";

		try {
			for (int i = 0; i < 50000; ++i) {
				File file = new File(folder + i + ".txt");
				boolean success = file.createNewFile();
				if (success) {
					;
				}
			}
		} catch (IOException e) {
		}
	}
}
